package raft

import (
	"bufio"
	"errors"
	"fmt"
	"io"
	"os"
	"sync"

	"github.com/goraft/raft/protobuf"
)

// ---------------------------------------------------------------
// 
// Typedefs 
// 
// ---------------------------------------

// /日志是持久存储的日志项集合。
type Log struct {
	ApplyFunc   func(*LogEntry, Command) (interface{}, error)
	file        *os.File
	path        string
	entries     []*LogEntry
	commitIndex uint64
	mutex       sync.RWMutex
	startIndex  uint64 // 日志项中第一个项之前的索引
	startTerm   uint64
	initialized bool
}

// 应用日志项的结果。
type logResult struct {
	returnValue interface{}
	err         error
}

// ---------------------------------------------------------------
// 
// 构造函数
// 
// ----------------------------------------------------------------

// 创建新日志。
func newLog() *Log {
	return &Log{
		entries: make([]*LogEntry, 0),
	}
}

// /-----------------------------------------------------------------
// /
// /访问器
// /
// /------------------------------------------------------------------

// /-----------------------------------------
// /日志索引
// /---------------------------------------

// 日志中最后提交的索引。
func (l *Log) CommitIndex() uint64 {
	l.mutex.RLock()
	defer l.mutex.RUnlock()
	return l.commitIndex
}

// 日志中的当前索引。
func (l *Log) currentIndex() uint64 {
	l.mutex.RLock()
	defer l.mutex.RUnlock()
	return l.internalCurrentIndex()
}

// 日志中的当前索引未锁定
func (l *Log) internalCurrentIndex() uint64 {
	if len(l.entries) == 0 {
		return l.startIndex
	}
	return l.entries[len(l.entries)-1].Index()
}

// 日志中的下一个索引。
func (l *Log) nextIndex() uint64 {
	return l.currentIndex() + 1
}

// 确定日志是否包含零项。
func (l *Log) isEmpty() bool {
	l.mutex.RLock()
	defer l.mutex.RUnlock()
	return (len(l.entries) == 0) && (l.startIndex == 0)
}

// 日志中最后一个命令的名称。
func (l *Log) lastCommandName() string {
	l.mutex.RLock()
	defer l.mutex.RUnlock()
	if len(l.entries) > 0 {
		if entry := l.entries[len(l.entries)-1]; entry != nil {
			return entry.CommandName()
		}
	}
	return ""
}

// /---------------------------------------
// 日志条款
// /---------------------------------------

// 日志中的当前条款。
func (l *Log) currentTerm() uint64 {
	l.mutex.RLock()
	defer l.mutex.RUnlock()

	if len(l.entries) == 0 {
		return l.startTerm
	}
	return l.entries[len(l.entries)-1].Term()
}

// ------------------------------------------------------------------
// /
// 方法
// /
// -------------------------------------------------------------

// ------------------------------------
// 状态
// --------------------------------------

// 打开日志文件并读取现有条目。日志可以保持打开状态，
// 继续在日志末尾追加条目。
func (l *Log) open(path string) error {
	// 读取日志中的所有条目（如果存在）。
	var readBytes int64

	var err error
	debugln("log.open.open ", path)
	// 打开日志文件
	l.file, err = os.OpenFile(path, os.O_RDWR, 0600)
	l.path = path

	if err != nil {
		// 如果日志文件在
		// 之前不存在，我们将创建日志文件并将commitIndex设置为0 
		if os.IsNotExist(err) {
			l.file, err = os.OpenFile(path, os.O_WRONLY|os.O_CREATE, 0600)
			debugln("log.open.create ", path)
			if err == nil {
				l.initialized = true
			}
			return err
		}
		return err
	}
	debugln("log.open.exist ", path)

	// 读取文件并解码条目。
	for {
		// 实例化日志条目并解码到其中。
		entry, _ := newLogEntry(l, nil, 0, 0, nil)
		entry.Position, _ = l.file.Seek(0, os.SEEK_CUR)

		n, err := entry.Decode(l.file)
		if err != nil {
			if err == io.EOF {
				debugln("open.log.append: finish ")
			} else {
				if err = os.Truncate(path, readBytes); err != nil {
					return fmt.Errorf("raft.Log: Unable to recover: %v", err)
				}
			}
			break
		}
		if entry.Index() > l.startIndex {
			// 追加条目。
			l.entries = append(l.entries, entry)
			if entry.Index() <= l.commitIndex {
				command, err := newCommand(entry.CommandName(), entry.Command())
				if err != nil {
					continue
				}
				l.ApplyFunc(entry, command)
			}
			debugln("open.log.append log index ", entry.Index())
		}

		readBytes += int64(n)
	}
	debugln("open.log.recovery number of log ", len(l.entries))
	l.initialized = true
	return nil
}

// 关闭日志文件。
func (l *Log) close() {
	l.mutex.Lock()
	defer l.mutex.Unlock()

	if l.file != nil {
		l.file.Close()
		l.file = nil
	}
	l.entries = make([]*LogEntry, 0)
}

// 同步到磁盘
func (l *Log) sync() error {
	return l.file.Sync()
}

// /---------------------------------------
// 条目
// /---------------------------------------

// 创建与此日志关联的日志条目。
func (l *Log) createEntry(term uint64, command Command, e *ev) (*LogEntry, error) {
	return newLogEntry(l, e, l.nextIndex(), term, command)
}

// 从日志中检索一个条目。如果由于快照的
// 而删除了该条目，则返回nil。
func (l *Log) getEntry(index uint64) *LogEntry {
	l.mutex.RLock()
	defer l.mutex.RUnlock()

	if index <= l.startIndex || index > (l.startIndex+uint64(len(l.entries))) {
		return nil
	}
	return l.entries[index-l.startIndex-1]
}

// 检查日志是否包含给定的索引/术语组合。
func (l *Log) containsEntry(index uint64, term uint64) bool {
	entry := l.getEntry(index)
	return (entry != nil && entry.Term() == term)
}

// 检索给定索引后的条目列表以及提供的
// 索引项。如果由于创建了快照，索引
// 不再存在，则返回nil条目列表。
func (l *Log) getEntriesAfter(index uint64, maxLogEntriesPerRequest uint64) ([]*LogEntry, uint64) {
	l.mutex.RLock()
	defer l.mutex.RUnlock()

	// 如果索引在日志开始之前，则返回nil。
	if index < l.startIndex {
		traceln("log.entriesAfter.before: ", index, " ", l.startIndex)
		return nil, 0
	}

	// 如果索引不存在，则返回错误。
	if index > (uint64(len(l.entries)) + l.startIndex) {
		panic(fmt.Sprintf("raft: Index is beyond end of log: %v %v", len(l.entries), index))
	}

	// 如果我们从日志的开头开始，那么返回整个日志。
	if index == l.startIndex {
		traceln("log.entriesAfter.beginning: ", index, " ", l.startIndex)
		return l.entries, l.startTerm
	}

	traceln("log.entriesAfter.partial: ", index, " ", l.entries[len(l.entries)-1].Index)

	entries := l.entries[index-l.startIndex:]
	length := len(entries)

	traceln("log.entriesAfter: startIndex:", l.startIndex, " length", len(l.entries))

	if uint64(length) < maxLogEntriesPerRequest {
		// 确定给定条目处的术语并返回子片段。
		return entries, l.entries[index-1-l.startIndex].Term()
	} else {
		return entries[:maxLogEntriesPerRequest], l.entries[index-1-l.startIndex].Term()
	}
}

// /-----------------------------------------
// /提交
// /------------------------------------

// 检索已提交到日志的最后一个索引和术语。
func (l *Log) commitInfo() (index uint64, term uint64) {
	l.mutex.RLock()
	defer l.mutex.RUnlock()
	// 如果没有任何提交的条目，则只返回零。
	if l.commitIndex == 0 {
		return 0, 0
	}

	// 快照后没有新的提交日志
	if l.commitIndex == l.startIndex {
		return l.startIndex, l.startTerm
	}

	// 返回上次提交项的最后一个索引和术语。
	debugln("commitInfo.get.[", l.commitIndex, "/", l.startIndex, "]")
	entry := l.entries[l.commitIndex-1-l.startIndex]
	return entry.Index(), entry.Term()
}

// 检索已附加到日志的最后一个索引和术语。
func (l *Log) lastInfo() (index uint64, term uint64) {
	l.mutex.RLock()
	defer l.mutex.RUnlock()

	// 如果没有任何条目，则只返回零。
	if len(l.entries) == 0 {
		return l.startIndex, l.startTerm
	}

	// 返回最后一个索引&term 
	entry := l.entries[len(l.entries)-1]
	return entry.Index(), entry.Term()
}

// 更新提交索引
func (l *Log) updateCommitIndex(index uint64) {
	l.mutex.Lock()
	defer l.mutex.Unlock()
	if index > l.commitIndex {
		l.commitIndex = index
	}
	debugln("update.commit.index ", index)
}

// 更新提交索引并将该索引后的项写入稳定存储器。
func (l *Log) setCommitIndex(index uint64) error {
	l.mutex.Lock()
	defer l.mutex.Unlock()

	// 在限制发送条目的数量后，这不再是错误
	// 提交到我们已经拥有的内容
	if index > l.startIndex+uint64(len(l.entries)) {
		debugln("raft.Log: Commit index", index, "set back to ", len(l.entries))
		index = l.startIndex + uint64(len(l.entries))
	}

	// 不允许再次提交以前的索引。

	// 这可能发生，因为保证新的领导者拥有最新的
	// 日志条目，而不是最新的提交索引

	// 例如，领导者1将日志80发送给追随者2和追随者3 
	// follower 2和follower 3都获得了新条目并回复
	// leader 1提交条目80并向follower 2和follower 3发送回复
	// follower 2收到新提交的索引并将提交的索引更新为80 
	// leader 1未能向follower 3发送提交的索引
	// 跟随者3升级为领导者（服务器1和服务器2将投票，因为领导者3 
	// 更新了条目）
	// 当新的领导者3向跟随者2发送提交索引为0的心跳信号时，
	// 关注者2应回复成功，并让领导者3将提交的索引更新为80 

	if index < l.commitIndex {
		return nil
	}

	// 查找索引位于上一个索引和当前索引之间的所有条目。
	for i := l.commitIndex + 1; i <= index; i++ {
		entryIndex := i - 1 - l.startIndex
		entry := l.entries[entryIndex]

		// 更新提交索引。
		l.commitIndex = entry.Index()

		// 解码命令。
		command, err := newCommand(entry.CommandName(), entry.Command())
		if err != nil {
			return err
		}

		// 将更改应用于状态机并存储错误代码。
		returnValue, err := l.ApplyFunc(entry, command)

		debugf("setCommitIndex.set.result index: %v, entries index: %v", i, entryIndex)
		if entry.event != nil {
			entry.event.returnValue = returnValue
			entry.event.c <- err
		}

		_, isJoinCommand := command.(JoinCommand)

		// 如果这批命令中有join，我们最多只能提交到最近的join命令
		// 中。
		// 在这次提交之后，我们需要重新计算多数。
		if isJoinCommand {
			return nil
		}
	}
	return nil
}

// 将日志文件头部的commitIndex设置为当前的
// 提交索引。获取日志锁后应调用此函数
func (l *Log) flushCommitIndex() {
	l.file.Seek(0, os.SEEK_SET)
	fmt.Fprintf(l.file, "%8x\n", l.commitIndex)
	l.file.Seek(0, os.SEEK_END)
}

// /---------------------------------------
// 截断
// /---------------------------------------

// 将日志截断为给定的索引和术语。这仅在索引处的日志
// 尚未提交时有效。
func (l *Log) truncate(index uint64, term uint64) error {
	l.mutex.Lock()
	defer l.mutex.Unlock()
	debugln("log.truncate: ", index)

	// 不允许截断提交的条目。
	if index < l.commitIndex {
		debugln("log.truncate.before")
		return fmt.Errorf("raft.Log: Index is already committed (%v): (IDX=%v, TERM=%v)", l.commitIndex, index, term)
	}

	// 不要截断过去的条目结尾。
	if index > l.startIndex+uint64(len(l.entries)) {
		debugln("log.truncate.after")
		return fmt.Errorf("raft.Log: Entry index does not exist (MAX=%v): (IDX=%v, TERM=%v)", len(l.entries), index, term)
	}

	// 如果要截断所有内容，只需清除条目即可。
	if index == l.startIndex {
		debugln("log.truncate.clear")
		l.file.Truncate(0)
		l.file.Seek(0, os.SEEK_SET)

		// 如果此节点是前一个领导者，请通知客户端
		for _, entry := range l.entries {
			if entry.event != nil {
				entry.event.c <- errors.New("command failed to be committed due to node failure")
			}
		}

		l.entries = []*LogEntry{}
	} else {
		// 如果索引中的项没有匹配项，请不要截断。
		entry := l.entries[index-l.startIndex-1]
		if len(l.entries) > 0 && entry.Term() != term {
			debugln("log.truncate.termMismatch")
			return fmt.Errorf("raft.Log: Entry at index does not have matching term (%v): (IDX=%v, TERM=%v)", entry.Term(), index, term)
		}

		// 否则截断到所需的条目。
		if index < l.startIndex+uint64(len(l.entries)) {
			debugln("log.truncate.finish")
			position := l.entries[index-l.startIndex].Position
			l.file.Truncate(position)
			l.file.Seek(position, os.SEEK_SET)

			// 如果此节点是前一个领导者，则通知客户端
			for i := index - l.startIndex; i < uint64(len(l.entries)); i++ {
				entry := l.entries[i]
				if entry.event != nil {
					entry.event.c <- errors.New("command failed to be committed due to node failure")
				}
			}

			l.entries = l.entries[0 : index-l.startIndex]
		}
	}

	return nil
}

// /-------------------------------------------------
// 追加
// /---------------------------------------

// 将一系列条目追加到日志中。
func (l *Log) appendEntries(entries []*protobuf.LogEntry) error {
	l.mutex.Lock()
	defer l.mutex.Unlock()

	startPosition, _ := l.file.Seek(0, os.SEEK_CUR)

	w := bufio.NewWriter(l.file)

	var size int64
	var err error
	// 追加每个条目，但如果遇到错误则退出。
	for i := range entries {
		logEntry := &LogEntry{
			log:      l,
			Position: startPosition,
			pb:       entries[i],
		}

		if size, err = l.writeEntry(logEntry, w); err != nil {
			return err
		}

		startPosition += size
	}
	w.Flush()
	err = l.sync()

	if err != nil {
		panic(err)
	}

	return nil
}

// 将单个日志条目写入日志末尾。
func (l *Log) appendEntry(entry *LogEntry) error {
	l.mutex.Lock()
	defer l.mutex.Unlock()

	if l.file == nil {
		return errors.New("raft.Log: Log is not open")
	}

	// 确保术语和索引大于上一个。
	if len(l.entries) > 0 {
		lastEntry := l.entries[len(l.entries)-1]
		if entry.Term() < lastEntry.Term() {
			return fmt.Errorf("raft.Log: Cannot append entry with earlier term (%x:%x <= %x:%x)", entry.Term(), entry.Index(), lastEntry.Term(), lastEntry.Index())
		} else if entry.Term() == lastEntry.Term() && entry.Index() <= lastEntry.Index() {
			return fmt.Errorf("raft.Log: Cannot append entry with earlier index in the same term (%x:%x <= %x:%x)", entry.Term(), entry.Index(), lastEntry.Term(), lastEntry.Index())
		}
	}

	position, _ := l.file.Seek(0, os.SEEK_CUR)

	entry.Position = position

	// 写入存储器。
	if _, err := entry.Encode(l.file); err != nil {
		return err
	}

	// 如果存储在磁盘上，则追加到条目列表。
	l.entries = append(l.entries, entry)

	return nil
}

// 带缓冲io的appendEntry 
func (l *Log) writeEntry(entry *LogEntry, w io.Writer) (int64, error) {
	if l.file == nil {
		return -1, errors.New("raft.Log: Log is not open")
	}

	// 确保术语和索引大于上一个。
	if len(l.entries) > 0 {
		lastEntry := l.entries[len(l.entries)-1]
		if entry.Term() < lastEntry.Term() {
			return -1, fmt.Errorf("raft.Log: Cannot append entry with earlier term (%x:%x <= %x:%x)", entry.Term(), entry.Index(), lastEntry.Term(), lastEntry.Index())
		} else if entry.Term() == lastEntry.Term() && entry.Index() <= lastEntry.Index() {
			return -1, fmt.Errorf("raft.Log: Cannot append entry with earlier index in the same term (%x:%x <= %x:%x)", entry.Term(), entry.Index(), lastEntry.Term(), lastEntry.Index())
		}
	}

	// 写入存储器。
	size, err := entry.Encode(w)
	if err != nil {
		return -1, err
	}

	// 如果存储在磁盘上，则追加到条目列表。
	l.entries = append(l.entries, entry)

	return int64(size), nil
}

// /---------------------------------------
// 日志压缩
// /---------------------------------------

// 在索引之前压缩日志（包括索引）
func (l *Log) compact(index uint64, term uint64) error {
	var entries []*LogEntry

	l.mutex.Lock()
	defer l.mutex.Unlock()

	if index == 0 {
		return nil
	}
	// 没有要压缩的内容
	// 如果
	// 我们刚刚从快照上恢复
	if index >= l.internalCurrentIndex() {
		entries = make([]*LogEntry, 0)
	} else {
		// 获取索引
		entries = l.entries[index-l.startIndex:]
	}

	// 创建新日志文件并添加所有条目
	new_file_path := l.path + ".new"
	file, err := os.OpenFile(new_file_path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600)
	if err != nil {
		return err
	}
	for _, entry := range entries {
		position, _ := l.file.Seek(0, os.SEEK_CUR)
		entry.Position = position

		if _, err = entry.Encode(file); err != nil {
			file.Close()
			os.Remove(new_file_path)
			return err
		}
	}
	file.Sync()

	old_file := l.file

	// 重命名新日志文件
	err = os.Rename(new_file_path, l.path)
	if err != nil {
		file.Close()
		os.Remove(new_file_path)
		return err
	}
	l.file = file

	// 关闭旧日志文件
	old_file.Close()

	// 压缩内存中的日志
	l.entries = entries
	l.startIndex = index
	l.startTerm = term
	return nil
}
